/*
    信道模块的实现：信道类、信道管理类
    1. 信道类：信道是在连接的基础上进一步细分，一个连接可以管理多个信道，而客户端的通信则是基于信道来完成的，一个信道能够提供所有的服务
        属性：信道标识、消费者句柄、protobuf协议处理器句柄、TcpConnection句柄、通用响应映射表<请求id, 通用响应对象>
        方法：打开/关闭信道、声明/删除交换机、声明/删除消息队列、绑定/解绑、发布、确认、订阅、取消订阅
    注1: 通用响应映射表用来存储当前信道上接收到的所有请求的响应，相当于存储异步结果的缓存
        因为发起请求和接收响应是在异步线程中完成的，所以需要异步通信，此处采用这种方式处理
    注2：当信道发起了订阅消息队列的请求时，才会创建一个消费者对象，消费者属性才有效
    注3：一个信道可以管理多个消费者，但是为了简化设计，这里一个信道只管理一个消费者
    注4: 信道的打开和关闭本应该在连接模块实现，但是在连接模块中实现较为麻烦，因为要处理发送打开信道请求之后的响应问题，而响应是异步线程的结果
        所以也就需要和信道模块处理通用响应方式一样，这样容易造成代码冗余，所以想服务端发起打开/关闭信道服务在此模块完成，本地创建信道对象依然在连接中完成
*/

#pragma once
#include "muduo/net/TcpConnection.h"
#include "proto/codec.h"
#include "consumer.hpp"
#include "../common/protocol.pb.h"
#include "../common/helper.hpp"
#include <mutex>
#include <condition_variable>

namespace mq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using CommonResponsePtr = std::shared_ptr<CommonResponse>;
    using ConsumeResponsePtr = std::shared_ptr<ConsumeResponse>;

    class Channel
    {
    public:
        using ChannelPtr = std::shared_ptr<Channel>;

    public:
        Channel(const muduo::net::TcpConnectionPtr &conn_ptr, const ProtobufCodecPtr &codec_ptr)
            : _id(UUIDHelper::uuid()), _conn_ptr(conn_ptr), _codec_ptr(codec_ptr)
        {}

        ~Channel()
        {
            // 信道关闭的时候，如果有订阅者应该取消订阅
            cancel(); 
        }

        bool openChannel()
        {
            // 1. 构建请求
            OpenChannelRequest req;
            std::string request_id = UUIDHelper::uuid();
            req.set_request_id(request_id);
            req.set_channel_id(_id);

            // 2. 发起响应
            _codec_ptr->send(_conn_ptr, req);

            // 3. 等待响应(阻塞等待)
            CommonResponsePtr rsp_ptr = waitResopnse(request_id);

            // 4. 返回结果
            return rsp_ptr->success();
        }

        // 关闭信道
        bool closeChannel()
        {
            // 1. 构建请求
            CloseChannelRequest req;
            std::string request_id = UUIDHelper::uuid();
            req.set_request_id(request_id);
            req.set_channel_id(_id);

            // 2. 发起响应
            _codec_ptr->send(_conn_ptr, req);

            // 3. 等待响应(阻塞等待)
            CommonResponsePtr rsp_ptr = waitResopnse(request_id);

            // 4. 返回结果
            return rsp_ptr->success();
        }

        // 声明交换机
        bool declareExchanger(const std::string& name, ExchangerType type, bool durable, bool auto_delete, const google::protobuf::Map<std::string, std::string>& args)
        {
            // 1. 构建请求
            DeclareExchangerRequest req;
            std::string request_id = UUIDHelper::uuid();
            req.set_request_id(request_id);
            req.set_channel_id(_id);
            req.set_exchanger_name(name);
            req.set_exchanger_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            auto* args_ptr = req.mutable_args();
            for(const auto& arg : args)
            {
                args_ptr->insert({arg.first, arg.second});
            }
            // DBG_LOG("declare exchanger request: request id: %s, name:%s, type:%d, durable:%d, auto_delete:%d\n", request_id.c_str(), name.c_str(), type, durable, auto_delete);
            // 2. 发起请求
            _codec_ptr->send(_conn_ptr, req);

            // 3. 等待响应(阻塞等待)
            CommonResponsePtr rsp_ptr = waitResopnse(request_id);

            // 4. 返回结果
            return rsp_ptr->success();
        }

        // 交换机的删除
        bool removeExchanger(const std::string &exchanger_name)
        {
            // 1. 构建请求
            RemoveExchangerRequest req;
            std::string request_id = UUIDHelper::uuid();
            req.set_request_id(request_id);
            req.set_channel_id(_id);
            req.set_exchanger_name(exchanger_name);

            // 2. 发起响应
            _codec_ptr->send(_conn_ptr, req);

            // 3. 等待响应(阻塞等待)
            CommonResponsePtr rsp_ptr = waitResopnse(request_id);

            // 4. 返回结果
            return rsp_ptr->success();
        }

        // 消息队列的声明
        bool declareMsgQueue(const std::string &name, bool durable, bool exclusive, bool auto_delete, const google::protobuf::Map<std::string, std::string>& args)
        {
            // 1. 构建请求
            DeclareQueueRequest req;
            std::string request_id = UUIDHelper::uuid();
            req.set_request_id(request_id);
            req.set_channel_id(_id);
            req.set_queue_name(name);
            req.set_durable(durable);
            req.set_exclusive(exclusive);
            auto* args_ptr = req.mutable_args();
            for(const auto& arg : args)
            {
                args_ptr->insert({arg.first, arg.second});
            }

            // 2. 发起响应
            _codec_ptr->send(_conn_ptr, req);

            // 3. 等待响应(阻塞等待)
            CommonResponsePtr rsp_ptr = waitResopnse(request_id);

            // 4. 返回结果
            return rsp_ptr->success();
        }

        // 删除消息队列
        bool removeMsgQueue(const std::string &name)
        {
            // 1. 构建请求
            RemoveQueueRequest req;
            std::string request_id = UUIDHelper::uuid();
            req.set_request_id(request_id);
            req.set_channel_id(_id);
            req.set_queue_name(name);

            // 2. 发起响应
            _codec_ptr->send(_conn_ptr, req);

            // 3. 等待响应(阻塞等待)
            CommonResponsePtr rsp_ptr = waitResopnse(request_id);

            // 4. 返回结果
            return rsp_ptr->success();
        }

        // 绑定
        bool bind(const std::string &exchanger_name, const std::string &msg_queue_name, const std::string &routing_key)
        {
            // 1. 构建请求
            BindRequest req;
            std::string request_id = UUIDHelper::uuid();
            req.set_request_id(request_id);
            req.set_channel_id(_id);
            req.set_exchanger_name(exchanger_name);
            req.set_queue_name(msg_queue_name);
            req.set_binding_key(routing_key);

            // 2. 发起响应
            _codec_ptr->send(_conn_ptr, req);
            // 3. 等待响应(阻塞等待)
            CommonResponsePtr rsp_ptr = waitResopnse(request_id);

            // 4. 返回结果
            return rsp_ptr->success();
        }

        // 解绑
        bool unbind(const std::string &exchanger_name, const std::string &msg_queue_name)
        {
            // 1. 构建请求
            BindRequest req;
            std::string request_id = UUIDHelper::uuid();
            req.set_request_id(request_id);
            req.set_channel_id(_id);
            req.set_exchanger_name(exchanger_name);
            req.set_queue_name(msg_queue_name);

            // 2. 发起响应
            _codec_ptr->send(_conn_ptr, req);

            // 3. 等待响应(阻塞等待)
            CommonResponsePtr rsp_ptr = waitResopnse(request_id);

            // 4. 返回结果
            return rsp_ptr->success();
        }

        // 消息的发布
        bool publish(const std::string &exchanger_name, const BasicProperties *properties, const std::string &body)
        {
            // 1. 构建请求
            PublishRequest req;
            std::string request_id = UUIDHelper::uuid();
            req.set_request_id(request_id);
            req.set_channel_id(_id);
            req.set_exchanger_name(exchanger_name);
            req.set_body(body);
            if(properties != nullptr)
            {
                req.mutable_properties()->set_id(properties->id());
                req.mutable_properties()->set_delivery_mode(properties->delivery_mode());
                req.mutable_properties()->set_routing_key(properties->routing_key());
            }

            // 2. 发起响应
            _codec_ptr->send(_conn_ptr, req);

            // 3. 等待响应(阻塞等待)
            CommonResponsePtr rsp_ptr = waitResopnse(request_id);

            // 4. 返回结果
            return rsp_ptr->success();
        }

        // 订阅队列消息
        bool consume(const std::string& consumer_tag, const std::string& queue_name, bool auto_ack, const Consumer::Callback& callback)
        {
            // 0. 判断当前信道是否已经订阅过
            if(_consumer_ptr != nullptr)
            {
                DBG_LOG("channel has already consumed message queue\n");
                return false;
            }

            // 1. 构建请求
            ConsumeRequest req;
            std::string request_id = UUIDHelper::uuid();
            req.set_request_id(request_id);
            req.set_channel_id(_id);
            req.set_queue_name(queue_name);
            req.set_consumer_tag(consumer_tag);
            req.set_auto_ack(auto_ack);

            // 2. 发起响应
            _codec_ptr->send(_conn_ptr, req);

            // 3. 等待响应(阻塞等待)
            CommonResponsePtr rsp_ptr = waitResopnse(request_id);

            // 4. 判断是否订阅成功，如果成功，则创建消费者对象，维护到当前信道
            if(rsp_ptr->success())
                _consumer_ptr = std::make_shared<Consumer>(consumer_tag, queue_name, auto_ack, callback);

            // 5. 返回结果
            return rsp_ptr->success();
        }

        // 取消订阅
        bool cancel()
        {
            // 0. 判断当前信道是否有订阅者
            if(_consumer_ptr == nullptr)
            {
                // DBG_LOG("channel has no consumer to cancel\n");
                return true;
            }

            // 1. 构建请求
            CancelRequest req;
            std::string request_id = UUIDHelper::uuid();
            req.set_request_id(request_id);
            req.set_channel_id(_id);
            req.set_consumer_tag(_consumer_ptr->_tag);
            req.set_queue_name(_consumer_ptr->_queue_name);

            // 2. 发起响应
            _codec_ptr->send(_conn_ptr, req);

            // 3. 等待响应(阻塞等待)
            CommonResponsePtr rsp_ptr = waitResopnse(request_id);

            // 4. 返回结果
            return rsp_ptr->success();
        }

        // 消息的确认
        bool ack(const std::string &msg_id)
        {
            // 0. 判断是否需要消息确认
            // (消息确认是对订阅的消息进行确认，如果当前信道没有消费者，则不需要确认，因为没有消费者就说明当前信道没有订阅过消息队列)
            if(_consumer_ptr == nullptr)
            {
                return true;
            }

            // 1. 构建请求
            AckRequest req;
            std::string request_id = UUIDHelper::uuid();
            req.set_request_id(request_id);
            req.set_channel_id(_id);
            req.set_queue_name(_consumer_ptr->_queue_name);
            req.set_message_id(msg_id);

            // 2. 发起响应
            _codec_ptr->send(_conn_ptr, req);

            // 3. 等待响应(阻塞等待)
            CommonResponsePtr rsp_ptr = waitResopnse(request_id);

            // 4. 返回结果
            return rsp_ptr->success();
        }

    private:
        // 从缓存responses中获取通用响应结果
        CommonResponsePtr waitResopnse(const std::string& request_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cond_var.wait(lock, [this, &request_id]() {
                return _responses.find(request_id) != _responses.end();
            });
            CommonResponsePtr resp_ptr = _responses[request_id];
            // DBG_LOG("wait response done, response id: %s\n", resp_ptr->response_id().c_str());
            _responses.erase(request_id);
            return resp_ptr;
        }
    
    public:
        /*
            下面是两种IO事件处理回调用函数：接收通用响应和接收消息推送的调用接口，最终由异步线程来执行
        */
        // 通用响应处理函数：接收到通用响应后，添加响应到responses异步缓存中
        void putCommonResponse(const CommonResponsePtr &rsp)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _responses.insert({rsp->response_id(), rsp});
            _cond_var.notify_one(); // 唤醒获取响应的线程
        }

        // 消息推送处理函数：接收到消息推送时，执行消费者的回调函数进行处理
        void consumeTask(const ConsumeResponsePtr &rsp)
        {
            // 1. 判断消费者是否存在
            if(_consumer_ptr == nullptr)
            {
                DBG_LOG("channel has no consumer, failed to push message\n");
                return;
            }

            // 2. 判断推送到消息是否属于当前消费者
            if(rsp->consumer_tag() != _consumer_ptr->_tag)
            {
                DBG_LOG("message is not belong to current consumer, failed to push message\n");
                return;
            }

            // 3. 执行订阅者回调
            _consumer_ptr->_callback(rsp->consumer_tag(), &rsp->properties(), rsp->body());
        }
    
    public:
        // 获取信道id
        std::string getChannelId() const
        {
            return _id;
        }

    private:
        std::string _id;                                                  // 信道标识
        Consumer::ConsumerPtr _consumer_ptr;                              // 消费者
        muduo::net::TcpConnectionPtr _conn_ptr;                           // 连接句柄
        ProtobufCodecPtr _codec_ptr;                                      // 序列化协议句柄
        std::mutex _mutex;                                                // 互斥锁
        std::condition_variable _cond_var;                                // 条件变量
        std::unordered_map<std::string, CommonResponsePtr> _responses;    // 通用响应消息映射表<请求id, 通用响应对象>
    };

    // 信道管理类
    class ChannelManager
    {
    public:
        using ChannelManagerPtr = std::shared_ptr<ChannelManager>;

    public:
        // 1. 创建信道
        Channel::ChannelPtr createChannel(const muduo::net::TcpConnectionPtr &conn_ptr, const ProtobufCodecPtr &codec_ptr)
        {
            // 这里信道创建不需要判断信道是否存在，因为信道id是在构造是随机生成的
            std::unique_lock<std::mutex> lock(_mutex);
            Channel::ChannelPtr channel_ptr = std::make_shared<Channel>(conn_ptr, codec_ptr);
            _channels.insert({channel_ptr->getChannelId(), channel_ptr});
            return channel_ptr;
        }

        // 2. 获取信道
        Channel::ChannelPtr getChannel(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(id);
            if (it != _channels.end())
                return it->second;
            return nullptr;
        }

        // 3. 删除信道
        void removeChannel(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(id);
        }

    private:
        std::unordered_map<std::string, Channel::ChannelPtr> _channels; // 信道映射表<信道id, 信道对象>
        std::mutex _mutex;
    };
}